package com.hkbigdata.watermark;

import com.hkbigdata.bean.WaterSensor;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @author liuanbo
 * @creat 2024-05-11-17:03
 * @see 2194550857@qq.com
 */
public class Flink05_SideOutPut {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> waterSensorSingleOutputStreamOperator = env.socketTextStream("localhost", 9999).flatMap(new FlatMapFunction<String, WaterSensor>() {
            @Override
            public void flatMap(String value, Collector<WaterSensor> out) throws Exception {
                String[] split = value.split(",");
                out.collect(new WaterSensor(split[0],
                        Long.valueOf(split[1]),
                        Double.valueOf(split[2])
                ));
            }
        });

        SingleOutputStreamOperator<String> process = waterSensorSingleOutputStreamOperator.keyBy(WaterSensor::getId)
                .process(new KeyedProcessFunction<String, WaterSensor, String>() {
                    @Override
                    public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {
                        //当传感器器大于45，报警
                        if (value.getVc() > 45) {
                            ctx.output(new OutputTag<String>("side") {
                            }, "传感器值超过" + value.getVc());
                        } else {
                            out.collect("传感器值没有超过：" + value.getVc());
                        }
                    }
                });

        process.print("主流");
        process.getSideOutput(new OutputTag<String>("side"){}).print("侧输出流");

        env.execute();

    }
}
